package org.nbict.iot.task.transit;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.kafka.spout.KafkaSpoutConfig;
import org.apache.storm.kafka.spout.trident.KafkaTridentSpoutOpaque;
import org.apache.storm.kafka.trident.TridentKafkaStateFactory;
import org.apache.storm.kafka.trident.TridentKafkaStateUpdater;
import org.apache.storm.kafka.trident.mapper.FieldNameBasedTupleToKafkaMapper;
import org.apache.storm.trident.Stream;
import org.apache.storm.trident.TridentTopology;
import org.apache.storm.trident.operation.BaseFilter;
import org.apache.storm.trident.state.StateFactory;
import org.apache.storm.trident.tuple.TridentTuple;
import org.apache.storm.tuple.Fields;
import org.nbict.iot.trident.kafka.FieldNameTopicSelector;
import org.nbict.iot.trident.kafka.KafkaSpout4RawTools;

import java.util.Properties;

/**
 *
 * EARLIEST - 0
 * LATEST - 1
 * UNCOMMITTED_EARLIEST - 2
 * UNCOMMITTED_LATEST - 3
 * Created by songseven on 18/7/6.
 */
public class PacketParserTrident {
    
    public static void main(String[] args) throws InvalidTopologyException, AuthorizationException, AlreadyAliveException {

        //设置kafka消费策略
        KafkaSpoutConfig.FirstPollOffsetStrategy strategy = KafkaSpoutConfig
                .FirstPollOffsetStrategy.EARLIEST;
        if (args.length > 1) {
            strategy = KafkaSpoutConfig.FirstPollOffsetStrategy.valueOf
                    (args[1]); //strategy name
        }

        TridentTopology topology = new TridentTopology();
        KafkaSpoutConfig spoutConfig = KafkaSpout4RawTools.newKafkaSpoutConfig
                ("192.168.1.27:9092", "raw-.*-simple", // bootstrap server raw-.*-simple 192.168.1.27:9092
                        "iot-car-raw-parser-group", strategy); // group id ignored
        KafkaTridentSpoutOpaque kafkaSpout = KafkaSpout4RawTools
                .newKafkaTridentSpoutOpaque(spoutConfig);

        Stream stream = topology.newStream("iot-car-raw", kafkaSpout);
        Stream parsedStream = stream.each(new Fields("platform_id", "raw"),new BinaryParser(), new Fields("transit_id","parsed", "correct"))
                .project(new Fields("transit_id", "parsed", "correct"));//output

        //解析流输出到kafka
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.1.83:9092"); //192.168.1.83:9092
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("acks", "all");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        StateFactory kafkaStateFactory = new TridentKafkaStateFactory()
                .withProducerProperties(props)
                .withKafkaTopicSelector(new FieldNameTopicSelector
                        ("transit_id", "transit")) //topic name default transit
                .withTridentTupleToKafkaMapper(new
                        FieldNameBasedTupleToKafkaMapper("transit_id", "parsed")); //persist field

//        StateFactory kafkaErrorFactory = new TridentKafkaStateFactory()
//                .withProducerProperties(props)
//                .withKafkaTopicSelector(new FieldNameTopicSelector(
//                        "transit_id", "error"))
//                .withTridentTupleToKafkaMapper(new FieldNameBasedTupleToKafkaMapper("transit_id", "raw"));

        //校验成功
        parsedStream.filter(new BaseFilter(){
            @Override
            public boolean isKeep(TridentTuple tuple) {
                return tuple.getBoolean(2);
            }
        }).partitionPersist(kafkaStateFactory, new Fields("transit_id","parsed"), //input field
                new TridentKafkaStateUpdater(), new Fields());

//        //校验失败
//        parsedStream.filter(new BaseFilter() {
//            @Override
//            public boolean isKeep(TridentTuple tuple) {
//                return !tuple.getBoolean(2);
//            }
//        }).partitionPersist(kafkaErrorFactory, new Fields("transit_id","raw"), //input field
//                new TridentKafkaStateUpdater(), new Fields());

        //解析错误
//        parsedStream.filter(new BaseFilter() {
//            @Override
//            public boolean isKeep(TridentTuple tuple) {
//                return !tuple.getBoolean(2);
//            }
//        }).partitionPersist(kafkaStateFactory, new Fields("transit_id","parsed"), //input field
//                new TridentKafkaStateUpdater(), new Fields());

        Config config = new Config();
        config.put("kafka.topic.wildcard.match", true);

        if (args.length > 0) {
            //集群提交模式
            config.setDebug(false);
            StormSubmitter.submitTopology(args[0], config, topology.build());
        } else {
            //本地测试模式
            config.setDebug(false);
            //设置1个进程
            config.setNumWorkers(1);
            LocalCluster cluster = new LocalCluster(); //new LocalCluster("127.0.0.1", 2181l);
            cluster.submitTopology("iot-car-parser-new", config, topology.build());
        }

    }
}
